/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.snda.dw.lidejia.hadooptotalsort.mr;

import com.snda.dw.lidejia.hadooptotalsort.type.SortKeys;
import java.io.IOException;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.util.LineReader;

/**
 *
 * @author lidejia
 */
public class SortKeysRecordReader implements RecordReader<SortKeys, Text> {

    private JobConf conf;
    private long start;
    private long end;
    private long pos;
    private CompressionCodecFactory compressionCodecs;
    private LineReader reader;
    private int maxLineLength;
    private SortKeysExtractor extractor;

    public SortKeysRecordReader(JobConf conf, FileSplit split) throws IOException {
        this.conf = conf;
        extractor = new SortKeysExtractor();
        //
        this.start = split.getStart();
        this.end = this.start + split.getLength();
        this.maxLineLength = conf.getInt("mapred.linerecordreader.maxlength",
                Integer.MAX_VALUE);
        //
        final Path file = split.getPath();
        compressionCodecs = new CompressionCodecFactory(conf);
        final CompressionCodec codec = compressionCodecs.getCodec(file);

        // open the file and seek to the start of the split
        FileSystem fs = file.getFileSystem(conf);
        FSDataInputStream fileIn = fs.open(file);
        boolean skipFirstLine = false;
        if (codec != null) {
            reader = new LineReader(codec.createInputStream(fileIn), conf);
            end = Long.MAX_VALUE;
        } else {
            if (start != 0) {
                skipFirstLine = true;
                --start;
                fileIn.seek(start);
            }
            reader = new LineReader(fileIn, conf);
        }
        if (skipFirstLine) {  // skip first line and re-establish "start".
            start += reader.readLine(new Text(), 0,
                    (int) Math.min((long) Integer.MAX_VALUE, end - start));
        }
        this.pos = start;
    }

    @Override
    public boolean next(SortKeys key, Text value) throws IOException {
        while (pos < end) {
            int newSize = reader.readLine(value, maxLineLength,
                    Math.max((int) Math.min(Integer.MAX_VALUE, end - pos),
                    maxLineLength));
            if (newSize == 0) {
                return false;
            }
            pos += newSize;
            if (newSize < maxLineLength) {
                key.setKeys(extractor.extract(value.toString()));
                return true;
            }
        }
        return false;
    }

    @Override
    public SortKeys createKey() {
        return new SortKeys();
    }

    @Override
    public Text createValue() {
        return new Text();
    }

    @Override
    public long getPos() throws IOException {
        return pos;
    }

    @Override
    public void close() throws IOException {
        if (reader != null) {
            reader.close();
        }
    }

    @Override
    public float getProgress() throws IOException {
        if (start == end) {
            return 0.0f;
        } else {
            return Math.min(1.0f, (pos - start) / (float) (end - start));
        }
    }
}
